package com.sg.java.util;

import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Predicate;

public class HdfsUtils {

    private static final Logger log = LoggerFactory.getLogger(HdfsUtils.class);

    public static FileSystem fs;

    public static final String hdfsPathPrefix = "/user/cq_qkjdyjc";

    public static final String uriPathPrefix = "hdfs://hacluster" + hdfsPathPrefix;

    static {
        try {
            String path_hdfs = ClassLoader.getSystemResource("hdfs-site.xml").getPath();
            String path_core = ClassLoader.getSystemResource("core-site.xml").getPath();
            log.info("path_hdfs:{}", path_hdfs);
            log.info("path_core:{}", path_core);
            Configuration conf = new Configuration();
            conf.addResource(new Path(path_hdfs));
            conf.addResource(new Path(path_core));
            conf.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
            conf.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
            UserGroupInformation.setConfiguration(conf);
            UserGroupInformation.loginUserFromSubject(null);
            fs = FileSystem.get(conf);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static Map<String, Pair<Map<String, Object>, List<String>>> read(String dirStr, Predicate<FileStatus> filter) throws IOException {
        if (!dirStr.contains(hdfsPathPrefix)) {
            dirStr = hdfsPathPrefix + "/" + dirStr;
        }
        Path                                                 dirPath = new Path(dirStr);
        FileStatus[]                                         fsArray = fs.listStatus(dirPath);
        Map<String, Pair<Map<String, Object>, List<String>>> result  = new ConcurrentHashMap<>();
        Arrays.stream(fsArray).filter(filter).parallel().map(dir -> {
            try {
                assert dir.isDirectory();
                return Pair.of(dir, fs.listStatus(dir.getPath()));
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }).forEach(pair -> {
            FileStatus          dir           = pair.getLeft();
            String              name          = dir.getPath().getName();
            FileStatus[]        files         = pair.getRight();
            long                totalLength   = dir.getLen();
            int                 capacity      = (int) totalLength / 1024;
            Map<String, Object> info          = new HashMap<>();
            int                 failFileCount = 0;
            List<String>        totalList     = new ArrayList<>(capacity);
            StringJoiner        error         = new StringJoiner(",");
            long                startTime     = System.currentTimeMillis();
            for (FileStatus file : files) {
                List<String> localList = new ArrayList<>(capacity / files.length);
                try (
                        FSDataInputStream is = fs.open(file);
                        InputStreamReader isr = new InputStreamReader(is);
                        BufferedReader br = new BufferedReader(isr)) {
                    String line;
                    while ((line = br.readLine()) != null) {
                        localList.add(line);
                    }
                } catch (Exception e) {
                    failFileCount++;
                    localList.clear();
                    error.add(e.getMessage());
                    break;
                }
                totalList.addAll(localList);
            }
            long endTime = System.currentTimeMillis() - startTime;
            info.put("总文件数", files.length);
            info.put("失败读取文件数", failFileCount);
            info.put("失败读取文件名", error.toString());
            info.put("数据总行数", totalList.size());
            info.put("总耗时", endTime);
            result.put(name, Pair.of(info, totalList));
        });
        return result;
    }

}
